package com.im.business.api;

import com.alibaba.fastjson.JSONObject;
import com.im.business.api.entity.GroupUser;
import com.im.business.api.entity.MessageReceive;
import com.im.business.api.entity.MessageSend;
import com.im.business.api.mapper.GroupUserMapper;
import com.im.business.api.mapper.MessageReceiveMapper;
import com.im.business.api.mapper.MessageSendMapper;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;

/**
 * @Created leijun
 * @Date 2021/10/16 8:28 上午
 */
public class Application {

    private static SqlSessionFactory sqlSessionFactory;

    public static void main(String[] args) throws IOException, InterruptedException {
        DataSourceManager dataSourceManager = DataSourceManager.getInstance();
        dataSourceManager.init();
        sqlSessionFactory = dataSourceManager.getSqlSessionFactory();

        KafkaManager kafkaManager = KafkaManager.getInstance();
        KafkaConsumer<String, String> consumer = kafkaManager.getConsumer();
        consumer.subscribe(Arrays.asList("single_message", "group_message", "push_message_response"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord record : records) {
                String topic = record.topic();
                System.out.println("【业务层】接收到kafka topic【"+topic+"】消息："+record.value());
                JSONObject message = JSONObject.parseObject(record.value().toString());
                if (topic.equals("single_message")) {
                    Long messageId = saveMessage(message);
                    sendMessageResponse(message, messageId);
                    pushMessage(message, messageId);
                } else if (topic.equals("group_message")) {
                    Long messageId = saveGroupMessage(message);
                    sendGroupMessageResponse(message, messageId);
                    pushGroupMessage(message, messageId);
                } else if (topic.equals("push_message_response")) {
                    Long messageId = message.getLong("messageId");
                    updateMessageDelivered(messageId);
                }
            }
            Thread.sleep(1000);
        }
    }

    /**
     * 消息保存到数据库
     * @param message
     * @return
     */
    private static Long saveMessage(JSONObject message) {
        SqlSession sqlSession = sqlSessionFactory.openSession();
        try {
            MessageSendMapper messageSendMapper = sqlSession.getMapper(MessageSendMapper.class);
            long messageId = insertMessageSend(messageSendMapper, message);
            MessageReceiveMapper messageReceiveMapper = sqlSession.getMapper(MessageReceiveMapper.class);
            insertMessageReceiver(messageReceiveMapper, message, messageId);
            return messageId;
        } finally {
            sqlSession.commit();
            sqlSession.close();
        }
    }

    /**
     * 保存群消息
     * @param message
     * @return
     */
    private static Long saveGroupMessage(JSONObject message) {
        SqlSession sqlSession = sqlSessionFactory.openSession();
        try {
            MessageSendMapper messageSendMapper = sqlSession.getMapper(MessageSendMapper.class);
            long messageId = insertMessageSend(messageSendMapper, message);
            MessageReceiveMapper messageReceiveMapper = sqlSession.getMapper(MessageReceiveMapper.class);
            // 查询群成员，给每个群成员都插入一条消息
            GroupUserMapper groupUserMapper = sqlSession.getMapper(GroupUserMapper.class);
            List<GroupUser> groupUsers = groupUserMapper.selectByGroup(message.getLong("groupId"));
            for (GroupUser user : groupUsers) {
                insertGroupMessageReceiver(messageReceiveMapper, message, messageId, user.getUserId());
            }
            return messageId;
        } finally {
            sqlSession.commit();
            sqlSession.close();
        }
    }

    /**
     * 新增发送消息
     * @param messageSendMapper
     * @param message
     * @return
     */
    public static Long insertMessageSend(MessageSendMapper messageSendMapper, JSONObject message) {
        MessageSend messageSend = new MessageSend();
        messageSend.setSenderId(message.getString("senderId"));
        messageSend.setReceiverId(message.getString("receiverId"));
        messageSend.setGroupId(message.getLong("groupId"));
        messageSend.setContent(message.getString("content"));
        messageSend.setRequestType(message.getInteger("requestType"));
        messageSend.setMessageType(message.getInteger("messageType"));
        messageSend.setSequence(message.getInteger("sequence"));
        messageSend.setSendTime(message.getLong("timestamp"));
        messageSendMapper.insert(messageSend);
        return messageSend.getId();
    }

    /**
     * 新增接收消息
     * @param messageReceiveMapper
     * @param message
     * @param messageId
     */
    private static void insertMessageReceiver(MessageReceiveMapper messageReceiveMapper,
                                              JSONObject message, long messageId) {
        MessageReceive messageReceive = new MessageReceive();
        messageReceive.setId(messageId);
        messageReceive.setSenderId(message.getString("senderId"));
        messageReceive.setReceiverId(message.getString("receiverId"));
        messageReceive.setContent(message.getString("content"));
        messageReceive.setRequestType(message.getInteger("requestType"));
        messageReceive.setMessageType(message.getInteger("messageType"));
        messageReceive.setSequence(message.getInteger("sequence"));
        messageReceive.setSendTime(message.getLong("timestamp"));
        messageReceive.setIsDelivered(0);
        messageReceiveMapper.insert(messageReceive);
    }

    /**
     * 新增群接收消息
     * @param messageReceiveMapper
     * @param message
     * @param messageId
     * @param receiverId
     */
    private static void insertGroupMessageReceiver(MessageReceiveMapper messageReceiveMapper,
                                              JSONObject message, long messageId, String receiverId) {
        MessageReceive messageReceive = new MessageReceive();
        messageReceive.setId(messageId);
        messageReceive.setSenderId(message.getString("senderId"));
        messageReceive.setReceiverId(receiverId);
        messageReceive.setGroupId(message.getLong("groupId"));
        messageReceive.setContent(message.getString("content"));
        messageReceive.setRequestType(message.getInteger("requestType"));
        messageReceive.setMessageType(message.getInteger("messageType"));
        messageReceive.setSequence(message.getInteger("sequence"));
        messageReceive.setSendTime(message.getLong("timestamp"));
        messageReceive.setIsDelivered(0);
        messageReceiveMapper.insert(messageReceive);
    }

    /**
     * 响应单聊消息
     * @param message
     * @param messageId
     */
    private static void sendMessageResponse(JSONObject message, Long messageId) {
        KafkaProducer producer = KafkaManager.getProducer();
        JSONObject messageResponse = new JSONObject();
        messageResponse.put("senderId", message.getString("senderId"));
        messageResponse.put("messageId", messageId);
        messageResponse.put("timestamp", message.getLong("timestamp"));
        ProducerRecord producerRecord = new ProducerRecord("single_message_response", messageResponse.toJSONString());
        producer.send(producerRecord);
    }

    /**
     * 响应群聊消息
     * @param message
     * @param messageId
     */
    private static void sendGroupMessageResponse(JSONObject message, Long messageId) {
        KafkaProducer producer = KafkaManager.getProducer();
        JSONObject messageResponse = new JSONObject();
        messageResponse.put("senderId", message.getString("senderId"));
        messageResponse.put("groupId", message.getLong("groupId"));
        messageResponse.put("messageId", messageId);
        messageResponse.put("timestamp", message.getLong("timestamp"));
        ProducerRecord producerRecord = new ProducerRecord("group_message_response", messageResponse.toJSONString());
        producer.send(producerRecord);
    }

    /**
     * 推送单聊消息
     * @param message
     * @param messageId
     */
    private static void pushMessage(JSONObject message, Long messageId) {
        KafkaProducer producer = KafkaManager.getProducer();
        JSONObject messageResponse = new JSONObject();
        messageResponse.put("senderId", message.getString("senderId"));
        messageResponse.put("receiverId", message.getString("receiverId"));
        messageResponse.put("messageId", messageId);
        messageResponse.put("content", message.getString("content"));
        messageResponse.put("timestamp", message.getLong("timestamp"));
        ProducerRecord producerRecord = new ProducerRecord("push_message", messageResponse.toJSONString());
        producer.send(producerRecord);
    }

    /**
     * 推送群聊消息
     * @param message
     * @param messageId
     */
    private static void pushGroupMessage(JSONObject message, Long messageId) {
        SqlSession sqlSession = sqlSessionFactory.openSession();
        try {
            Long groupId = message.getLong("groupId");
            GroupUserMapper groupUserMapper = sqlSession.getMapper(GroupUserMapper.class);
            List<GroupUser> groupUsers = groupUserMapper.selectByGroup(groupId);
            KafkaProducer producer = KafkaManager.getProducer();
            for (GroupUser groupUser : groupUsers) {
                JSONObject messageResponse = new JSONObject();
                messageResponse.put("senderId", message.getString("senderId"));
                messageResponse.put("receiverId", groupUser.getUserId());
                messageResponse.put("groupId", groupId);
                messageResponse.put("messageId", messageId);
                messageResponse.put("content", message.getString("content"));
                messageResponse.put("timestamp", message.getLong("timestamp"));
                ProducerRecord producerRecord = new ProducerRecord("push_group_message", messageResponse.toJSONString());
                producer.send(producerRecord);
            }
        }  finally {
            sqlSession.close();
        }
    }

    /**
     * 更新消息状态为已送达
     * @param messageId
     */
    private static void updateMessageDelivered(Long messageId) {
        SqlSession sqlSession = sqlSessionFactory.openSession();
        try {
            MessageReceiveMapper messageReceiveMapper = sqlSession.getMapper(MessageReceiveMapper.class);
            MessageReceive messageReceive = new MessageReceive();
            messageReceive.setId(messageId);
            messageReceive.setIsDelivered(1);
            messageReceive.setDeliverTime(System.currentTimeMillis());
            messageReceiveMapper.updateByPrimaryKey(messageReceive);
        } finally {
            sqlSession.commit();
            sqlSession.close();
        }
    }

}
